package com.example.reactor;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.Optional;
import java.util.concurrent.TimeUnit;

public class ReactorTest {

    @Test
    public void test01() {
        System.out.println("hello world");
        Flux.just(1, 2, 3, 4, 5, 6).subscribe(System.out::println);
        Mono.just(1);

        // 只有完成信号的空数据流
        Flux.just();
        Flux.empty();
        Mono.empty();
        Mono.justOrEmpty(Optional.empty());

        // 只有错误信号的数据流
        Flux.error(new Exception("some error"));
    }


    @Test
    public void test02() {
        Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9).map(num -> num * 3).subscribe(System.out::println);
    }

    @Test
    public void test03() throws InterruptedException {
        // 同步阻塞场景
        System.out.println(hello());
        System.out.println("-----------同步阻塞场景执行结束------------");
        helloAsync();
        System.out.println("-----------异步阻塞场景执行结束------------");
        Thread.sleep(3000);
    }

    private String hello() {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "hello, Reactor!";
    }

    private void helloAsync() {
        Mono.fromCallable(() -> hello())
                .subscribeOn(Schedulers.elastic())
                .subscribe(System.out::println, System.err::println);
    }

}
